/*
 * @Author: gonglf 457423258@qq.com
 * @Date: 2022-06-11 16:41:48
 * @LastEditors: gonglf
 * @LastEditTime: 2022-09-23 10:47:51
 * @FilePath: /consumer-demo/consumer.go
 * @Description:
 *
 * Copyright (c) 2022 by gonglf 457423258@qq.com, All Rights Reserved.
 */
package consumer

import (
	"context"
	"log"
	"runtime"
	"sync"
	"time"
)

var (
	// stopChan        chan struct{}
	consumingWG     sync.WaitGroup //正在消费任务
	processingWG    sync.WaitGroup //正在处理的任务
	concurrency     int            //并发数量，越大：chan等待被处理的越多，退出时需要更久，越小：并发数不多
	jobChan         chan []byte    //用来存放等待处理的任务数据，生产者
	workPoolProduce chan struct{}  //用来控制有多少个协程（生产者，消费者，worker）在工作，这个工作可以是：从redis中取出任务
	workPoolConsume chan struct{}  //用来控制有多少个协程（生产者，消费者，worker）在工作，这个工作可以是：worker消费任务
)

func init() {
	// stopChan = make(chan struct{})
	concurrency = 2
	if concurrency < 1 {
		concurrency = runtime.NumCPU() * 2
	}
	jobChan = make(chan []byte, concurrency)
	workPoolProduce = make(chan struct{}, concurrency)
	workPoolConsume = make(chan struct{}, concurrency)
}

func StartConsuming(ctx context.Context) (bool, error) { //形参传进一个接口，用来处理任务TODO

	consumingWG.Add(1)
	go func() {
		defer func() {
			defer consumingWG.Done()
		}()
		log.Println("Waiting for messages. To exit press CTRL+C")
		for {
			select {
			case <-ctx.Done():
				return
			case workPoolProduce <- struct{}{}: //上一把锁
				select {
				case <-ctx.Done():
					return
				default:
				}
				go func() { //这边为什么是一个协程？
					defer func() {
						<-workPoolProduce //归还锁
					}()
					//消费任务
					task, _ := getTask()
					if len(task) > 0 {
						processingWG.Add(1)
						log.Println("++++")
						jobChan <- task
					}
				}()
			}
		}
	}()

	go consume()

	return true, nil
}

//各种close
func StopConsuming() {
	consumingWG.Wait()
	processingWG.Wait()
	// close(jobChan) //任务消费完了才能close

}

func consume() {

	for d := range jobChan { //TODO 这边可以改为for select，退出时把未消费的放回redis，提高退出效率

		workPoolConsume <- struct{}{} //加一把锁，同时只能最多执行concurrency个任务

		go func(d []byte) {

			defer func() {
				processingWG.Done()
				<-workPoolConsume //释放一把锁
			}()

			log.Println("----")
			consumeOne(d)
		}(d)
	}
}

// 处理一条数据，通常比较耗时
func consumeOne(delivery []byte) error { //TODO 这边需要一个接口来处理任务
	defer func() { //防止业务处理发声panic
		if err := recover(); err != nil {
			log.Println("err = ", err)
		}
	}()
	//业务逻辑在这里处理
	log.Println("processing:", string(delivery))
	time.Sleep(time.Second * 3)
	return nil
}

func getTask() (result []byte, err error) {
	//这边用BLPOP，阻塞式获取数据
	time.Sleep(time.Second * 1)
	s := "task"
	return []byte(s), nil
}
